package consumer.handler;

import com.google.protobuf.ByteString;
import consumer.config.AppConfig;
import consumer.config.ZKConfig;
import consumer.enity.RegisterConfigOuterClass;
import consumer.strategy.RouterStrategy;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;

import java.util.*;

/**
 * @ClassName RouterHandler
 * @Description
 * @Author Jingxiao.Ye
 * @Date 2022/1/16 下午4:14
 * @Version V1.0
 * @Since JDK 11
 */
public class Router {

    Map<String,Node> consumerMap = new HashMap<>();

    List<String> consumerList = new ArrayList<>();

    RouterStrategy routerStrategy;

    public Router(RouterStrategy routerStrategy) {
        this.routerStrategy = routerStrategy;
        try {
            init();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    //注册
    public void init() throws Exception {
        //从zk中读取producer url
        String parentPath = "/" + AppConfig.group + "/" + AppConfig.version;
        List<String> list = ZKConfig.client.getChildren().forPath(parentPath);
        //获得其所有结点
        for (String s : list) {
            //获得节点的数据
            byte[] bytes = ZKConfig.client.getData().forPath(parentPath+"/"+s);
            RegisterConfigOuterClass.RegisterConfig registerConfig = RegisterConfigOuterClass.RegisterConfig.parseFrom(bytes);
            String host = registerConfig.getHost();
            int port = registerConfig.getPort();
            ByteString ext = registerConfig.getExt();
            Node node = new Node();
            node.path = s;
            node.url = host+":"+port;
            node.data = ext.toByteArray();
            consumerList.add(s);
            consumerMap.put(s,node);
        }
        //对节点监听变化
        PathChildrenCache pathChildrenCache = new PathChildrenCache(ZKConfig.client, parentPath, true);
        pathChildrenCache.getListenable().addListener((a,b)->{
            String path = b.getData().getPath();
            byte[] data = b.getData().getData();
            RegisterConfigOuterClass.RegisterConfig registerConfig = RegisterConfigOuterClass.RegisterConfig.parseFrom(data);
            String host = registerConfig.getHost();
            int port = registerConfig.getPort();
            ByteString ext = registerConfig.getExt();
            Node node = new Node();
            node.path = path;
            node.url = host+":"+port;
            node.data = ext.toByteArray();
            switch (b.getType()){
                case CHILD_ADDED:
                    if(consumerMap.get(path) == null){
                        consumerMap.put(path,node);
                        consumerList.add(path);
                    }
                    break;

                case CHILD_REMOVED:
                    consumerList.remove(path);
                    consumerMap.remove(path);
                    break;
            }
        });
    }

    public String getUrl(){
        if(consumerList.size() == 0) return null;
        //配置策略
        String router = routerStrategy.getRouter(consumerList);
        if(router == null){
            return null;
        }
        Node node = consumerMap.get(router);
        return node.getUrl();
    }

    public static class Node{
        private String path;

        private String url;

        private byte[] data;

        public String getPath() {
            return path;
        }

        public void setPath(String path) {
            this.path = path;
        }

        public String getUrl() {
            return url;
        }

        public void setUrl(String url) {
            this.url = url;
        }

        public byte[] getData() {
            return data;
        }

        public void setData(byte[] data) {
            this.data = data;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            Node node = (Node) o;
            return Objects.equals(path, node.path);
        }

        @Override
        public int hashCode() {
            return Objects.hash(path);
        }
    }
}
